package com.you.learning.rocketmq.producer;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

/**
 * @Description 生产消息到RocketMQ
 * @Author yough
 * @Version V1.0.0
 * @Since 1.0
 * @Date 2021/3/23
 */
public class ProducerService {
    public static void main(String[] args) throws Exception {
        //组名，在一个应用中必须唯一，如果两个相同名称则会报错。
        DefaultMQProducer producer = new DefaultMQProducer("finance");
        // 连接的NameServer地址
        producer.setNamesrvAddr("112.124.46.106:9876");
        // 实例名称, 使用了工厂模式， 公用一个i nstance 实例。多集群连接的问题，需要一个新的 instance 实例。
        producer.setInstanceName("ZaneProducerInstance");
        // 设置topic的队列数量
        producer.setDefaultTopicQueueNums(4);
        // 设置发送消息的超时时间
        producer.setSendMsgTimeout(3000);
        // 设置发送失败的重试次数
        producer.setRetryTimesWhenSendFailed(2);
        // 128k，设置消息的最大字节
        producer.setMaxMessageSize(128);
        producer.setVipChannelEnabled(false);
        // 启用消息生产者
        producer.start();
        /**
         * 同步发送消息
         */
        Message msg = new Message("test_topic_1", "TagA", ("Hello RocketMQ ").getBytes("UTF-8"));
        // 同步发送消息
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);
        /**
         * 异步发送消息
         */
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("%s%n", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {

            }
        });
        // 关闭消息生产者
        //producer.shutdown();
    }
}
